package com.leilei.count;

import cn.hutool.setting.dialect.Props;
import com.leilei.Location;
import com.leilei.sink.OverSpeedAlarmSinkFunction;
import com.leilei.sink.RabbitMqSinkProperties;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/14 21:55
 * @desc flink 窗口 根据数量开窗口
 */
public class FlinkWindow_CountKeyWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(4);
        DataStreamSource<Location> source = env.addSource(new MyLocationSource());
        WindowedStream<Location, Integer, GlobalWindow> keyWindow = source.keyBy(Location::getVehicleId)
                .countWindow(2);
        SingleOutputStreamOperator<String> result = keyWindow.apply(new CountKeyWindowFunction());
        result.print();
        env.execute("count-window");
    }


}

